package com.donglink.project.logmonitor.bolt;

import com.donglink.project.logmonitor.domain.LogMonitorRuleRecord;
import com.donglink.project.logmonitor.domain.MyMessage;
import com.donglink.project.logmonitor.utils.MonitorHandler;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IBasicBolt;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.IWindowedBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

public class NotifyBolt extends BaseRichBolt {
    OutputCollector outputCollector;
    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.outputCollector=outputCollector;
    }

    @Override
    public void execute(Tuple tuple) {
        //发邮件给所有需要通知的人，并传发送邮件结果和日志错误信息到下一个bolt
        MyMessage message=(MyMessage)tuple.getValue(0);
        LogMonitorRuleRecord logMonitorRuleRecord= MonitorHandler.sendMail(message);
        outputCollector.emit(new Values(logMonitorRuleRecord));


    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("notifylog"));
    }
}
